package main

import (
	"fmt"

	"context"
	"encoding/json"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/segmentio/kafka-go"
	"log"
	"saasems/services"
)

func connectMQTT(broker string, clientID string, username string, password string) mqtt.Client {
	opts := mqtt.NewClientOptions().AddBroker(broker).SetClientID(clientID).SetUsername(username).SetPassword(password)
	client := mqtt.NewClient(opts)
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		log.Printf("Error connecting to MQTT broker: %v", token.Error())
	}
	return client
}

func subscribeToDeviceTopic(client mqtt.Client, tenantID, gatewayID, deviceID string) {
	topics := []string{
		fmt.Sprintf("tenant/%s/gateway/%s/device/%s/register", tenantID, gatewayID, deviceID),
		fmt.Sprintf("tenant/%s/gateway/%s/device/%s/telemetry", tenantID, gatewayID, deviceID),
		fmt.Sprintf("tenant/%s/gateway/%s/device/%s/status", tenantID, gatewayID, deviceID),
		fmt.Sprintf("tenant/%s/gateway/%s/device/%s/control", tenantID, gatewayID, deviceID),
		fmt.Sprintf("tenant/%s/gateway/%s/device/%s/adjust", tenantID, gatewayID, deviceID),
	}

	for _, topic := range topics {
		if token := client.Subscribe(topic, 0, createMessageHandler(tenantID, gatewayID, deviceID)); token.Wait() && token.Error() != nil {
			log.Printf("Error subscribing to topic %s: %v", topic, token.Error())
		}
	}
}

func createMessageHandler(tenantID, gatewayID, deviceID string) mqtt.MessageHandler {
	return func(client mqtt.Client, msg mqtt.Message) {
		var data map[string]interface{}
		if err := json.Unmarshal(msg.Payload(), &data); err != nil {
			log.Printf("Error parsing message: %v", err)
			return
		}

		// 添加租户ID到数据中
		data["tenant_id"] = tenantID

		// 发送数据到 Kafka
		if err := sendToKafka(msg.Topic(), data); err != nil {
			log.Printf("Error sending to Kafka: %v", err)
		}
	}
}

func handleDeviceRegistration(data map[string]interface{}) {
	// 处理设备自注册逻辑
	// 将设备信息存储到数据库
	//db := database.GetDB()
	//// ... 数据库操作 ...
	//fmt.Println(db)
}

func handleTelemetryData(data map[string]interface{}) {
	// 处理遥测数据
	services.ProcessTelemetryData(data)
}

func handleStatusData(data map[string]interface{}) {
	// 处理遥信数据
	services.ProcessStatusData(data)
}

func handleControlData(data map[string]interface{}) {
	// 处理遥控数据
	services.ProcessControlData(data)
}

func handleAdjustmentData(data map[string]interface{}) {
	// 处理遥调数据
	services.ProcessAdjustmentData(data)
}

func sendToKafka(topic string, data map[string]interface{}) error {
	writer := &kafka.Writer{
		Addr:     kafka.TCP("localhost:9092"),
		Topic:    "mqtt-data",
		Balancer: &kafka.LeastBytes{},
	}

	jsonData, err := json.Marshal(data)
	if err != nil {
		return err
	}

	return writer.WriteMessages(context.Background(),
		kafka.Message{
			Key:   []byte(topic),
			Value: jsonData,
		},
	)
}
